package org.example;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class PaimonDDLCTAS {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT,"8081");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(conf);
        env.setParallelism(1);
        TableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);

        tableEnv.executeSql("CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'warehouse' = 'file:///tmp/paimon'\n" +
                ");\n");

        tableEnv.executeSql("DROP TABLE IF EXISTS `paimon`.`default`.`test`;");

        tableEnv.executeSql("CREATE TABLE IF NOT EXISTS `paimon`.`default`.`test` (\n" +
                "    user_id BIGINT NOT NULL,\n" +
                "    item_id BIGINT,\n" +
                "    behavior STRING,\n" +
                "    dt STRING NOT NULL,\n" +
                "    hh STRING NOT NULL,\n" +
                "    proc_time AS PROCTIME(),\n" +
                "    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED\n" +
                ");");

        tableEnv.executeSql("CREATE TEMPORARY TABLE temp_user_behavior(\n" +
                "    user_id BIGINT NOT NULL\n" +
                "    , item_id BIGINT\n" +
                "    , behavior STRING\n" +
                "    , dt STRING\n" +
                "    , hh STRING\n" +
                ")WITH (\n" +
                "  'connector' = 'filesystem',\n" +
                "  'path' = 'file:///tmp/test/temp_user_behavior.csv',\n" +
                "  'format' = 'csv'\n" +
                ");\n");

        tableEnv.executeSql("INSERT INTO `paimon`.`default`.`test`\n" +
                "SELECT\n" +
                "    user_id\n" +
                "    , item_id\n" +
                "    , behavior\n" +
                "    , dt\n" +
                "    , hh\n" +
                "FROM\n" +
                "    temp_user_behavior\n" +
                ";\n");

        tableEnv.executeSql("desc `paimon`.`default`.`test`").print();
    }
}
